package com.atguigu.flink.chapter07.state;

import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/8/16 15:03
 */
public class Flink02_Operator_BC {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);
        env.enableCheckpointing(3000);
        
        DataStreamSource<String> dataStream = env
            .socketTextStream("hadoop162", 8888);
        
        DataStreamSource<String> controlStream = env
            .socketTextStream("hadoop162", 9999);
        MapStateDescriptor<String, Integer> switchStateDesc = new MapStateDescriptor<>("switchState", String.class, Integer.class);
        
        // 1. 先把controlStream做成一个广播流   广播状态的本质: 就是一个Map集合   "switch"-> 1/2/3
        BroadcastStream<String> bcStream = controlStream
            .broadcast(switchStateDesc);
        
        // 2. 数据流和广播流进行connect
        dataStream
            .connect(bcStream)
            .process(new BroadcastProcessFunction<String, String, String>() {
                @Override
                public void processElement(String value,
                                           ReadOnlyContext ctx,
                                           Collector<String> out) throws Exception {
                    // 根据广播状态中不同的值, 执行不同的业务逻辑
                    ReadOnlyBroadcastState<String, Integer> bcState = ctx.getBroadcastState(switchStateDesc);
                    Integer logicNum = bcState.get("switch");
                    logicNum = logicNum == null ? -1 : logicNum;
                    if (logicNum.equals(1)) {
                        out.collect("使用 1号逻辑处理数据...");
                    } else if (logicNum.equals(2)) {
                        out.collect("使用 2号逻辑处理数据...");
                    } else {
                        out.collect("使用默认逻辑处理数据...");
                    }
                }
                
                @Override
                public void processBroadcastElement(String value,
                                                    Context ctx,
                                                    Collector<String> out) throws Exception {
                    // 根据广播流中的数据, 向广播状态中存储数据
                    Integer logicNum = Integer.valueOf(value);
                    
                    // 获取广播状态
                    BroadcastState<String, Integer> bcState = ctx.getBroadcastState(switchStateDesc);
                    bcState.put("switch", logicNum);
                    
                }
            })
            .print();
        
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
}
